package com.kingsoft.dc.khaos.module.spark.preprocess.specific.quality

import java.util.Properties

import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import scala.collection.mutable.ListBuffer

/**
  * 取值约束检查
  *
  *
  */
class ConstraintsValueCheck extends Logging with Serializable {
  def analysisData(data1: DataFrame, data2: DataFrame, expression1: String, expression2: String, rows: ListBuffer[Row]): ListBuffer[Row] = {
    //值获取
    val ex1 = expression1.split(" ")
    val ex2 = expression2.split(" ")
    val s1 = " " + ex1(1) + " " + ex1(2) + " " + ex1(3) + " "
    val str1: String = expression1.split(s1)(1)
    val s2 = " " + ex2(1) + " " + ex2(2) + " " + ex2(3) + " "
    val str2: String = expression2.split(s2)(1)
    val col1 = ex1(1).toLowerCase
    val col2 = ex2(1).toLowerCase
    //条件判断
    //对每一行进行遍历，判断后的得到的是符合要求的行
    val list1 = data1.collectAsList()
    val list2 = data2.collectAsList()
    try {
      for (i <- 0 to list1.size() - 1) {
        val x: Row = list1.get(i)
        val y: Row = list2.get(i)
        if (x.getAs(col1) != null && y.getAs(col2) != null) {
          ex1(3) match {
            case "=" => {
              val col1Vlue = x.getAs(col1).toString
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (str1.equals(col1Vlue) && !str2.equals(yString)) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (str1.equals(col1Vlue) && str2.equals(yString)) {
                    rows += y
                  }
                }
                case ">" => {
                  if (str1.equals(col1Vlue) && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (str1.equals(col1Vlue) && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (str1.equals(col1Vlue) && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (str1.equals(col1Vlue) && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (str1.equals(col1Vlue) && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            case "!=" => {
              val col1Vlue = x.getAs(col1).toString
              //打印的timestamp数据15:24:09
              val yString = y.getAs(col2).toString //打印的timestamp数据2019-05-08 15:24:09.0
              ex2(3) match {
                case "=" => {
                  if (!str1.equals(col1Vlue) && !str2.equals(yString)) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (!str1.equals(col1Vlue) && str2.equals(yString)) {
                    rows += y
                  }
                }
                case ">" => {
                  if (!str1.equals(col1Vlue) && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (!str1.equals(col1Vlue) && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (!str1.equals(col1Vlue) && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (!str1.equals(col1Vlue) && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (!str1.equals(col1Vlue) && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            //930新加需求
            case ">" => {
              val col1Vlue = x.getAs(col1).toString.toDouble
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble != yString.toDouble) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble == yString.toDouble) {
                    rows += y
                  }
                }
                case ">" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (str1.toDouble < col1Vlue && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (str1.toDouble < col1Vlue && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            case ">=" => {
              val col1Vlue = x.getAs(col1).toString.toDouble
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble != yString.toDouble) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble == yString.toDouble) {
                    rows += y
                  }
                }
                case ">" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (str1.toDouble <= col1Vlue && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (str1.toDouble <= col1Vlue && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            case "<" => {
              val col1Vlue = x.getAs(col1).toString.toDouble
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble != yString.toDouble) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble == yString.toDouble) {
                    rows += y
                  }
                }
                case ">" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (str1.toDouble > col1Vlue && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (str1.toDouble > col1Vlue && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            case "<=" => {
              val col1Vlue = x.getAs(col1).toString.toDouble
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble != yString.toDouble) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble == yString.toDouble) {
                    rows += y
                  }
                }
                case ">" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (str1.toDouble >= col1Vlue && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "in" => {
                  if (str1.toDouble >= col1Vlue && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
              }
            }
            case "in" => {
              val bool: Boolean = str1.split(",").contains(x.getAs(col1).toString)
              val yString = y.getAs(col2).toString
              ex2(3) match {
                case "=" => {
                  if (bool && !yString.equals(str2)) {
                    rows += y
                  }
                }
                case "!=" => {
                  if (bool && yString.equals(str2)) {
                    rows += y
                  }
                }
                case "in" => {
                  if (bool && !str2.split(",").contains(yString)) {
                    rows += y
                  }
                }
                case ">=" => {
                  if (bool && str2.toDouble > yString.toDouble) {
                    rows += y
                  }
                }
                case ">" => {
                  if (bool && str2.toDouble >= yString.toDouble) {
                    rows += y
                  }
                }
                case "<=" => {
                  if (bool && str2.toDouble < yString.toDouble) {
                    rows += y
                  }
                }
                case "<" => {
                  if (bool && str2.toDouble <= yString.toDouble) {
                    rows += y
                  }
                }
              }
            }
          }
        }
      }
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        println(ex)
      }
    }
    rows
  }
}


